package com.lingjtx.common.thread.pooling;

import com.lingjtx.common.thread.core.PoolNameConst;
import com.lingjtx.common.thread.core.ThreadPoolManager;
import com.lingjtx.common.thread.mq.FastJsonAutoTypeMessageConverter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;

import java.nio.charset.StandardCharsets;

public class RabbitTemplateFactory implements ObjectFactory<RabbitTemplate> {
    private static final Logger log = LoggerFactory.getLogger(RabbitTemplateFactory.class);

    private final ThreadPoolManager threadPoolManager;
    private final CachingConnectionFactory connectionFactory;
    private final FastJsonAutoTypeMessageConverter converter;

    public RabbitTemplateFactory(ThreadPoolManager threadPoolManager, CachingConnectionFactory connectionFactory, FastJsonAutoTypeMessageConverter converter) {
        this.threadPoolManager = threadPoolManager;
        this.connectionFactory = connectionFactory;
        this.converter = converter;
    }

    @Override
    public RabbitTemplate create() {
        connectionFactory.setExecutor(threadPoolManager.getOrCreate(PoolNameConst.RABBITMQ_POOL));
        RabbitTemplate template = new RabbitTemplate(connectionFactory);
        template.setMessageConverter(converter);
        template.setTaskExecutor(threadPoolManager.getOrCreate(PoolNameConst.RABBITMQ_POOL));

        // 启用 mandatory 模式以确保 ReturnCallback 能生效
        template.setMandatory(true);

        // 设置 ConfirmCallback：消息是否到达交换机
        template.setConfirmCallback((correlationData, ack, cause) -> {
            if (ack) {
                // 消息成功到达交换机
                log.info("✅ [ConfirmCallback] 消息发送成功: {}", correlationData);
            } else {
                // 消息未能到达交换机
                log.error("❌ [ConfirmCallback] 消息发送失败: {}, 原因: {}", correlationData, cause);
                if (correlationData != null) {
                    correlationData.getFuture().complete(new CorrelationData.Confirm(false, cause));
                }
            }
        });

        // 设置 ReturnCallback：消息未能从交换机路由到队列
        template.setReturnsCallback(returned -> {
            String body = new String(returned.getMessage().getBody(), StandardCharsets.UTF_8);
            log.warn("❗ [ReturnCallback] 消息未路由成功 -> 内容: {}, 原因: {}, 交换机: {}, 路由键: {}",
                    body, returned.getReplyText(), returned.getExchange(), returned.getRoutingKey());
        });
        return template;
    }

    @Override
    public boolean validate(RabbitTemplate obj) {
        return obj != null;
    }

    @Override
    public void destroy(RabbitTemplate obj) {
        shutdown();
        if (connectionFactory != null) {
            try {
                connectionFactory.destroy();
            } catch (Exception ignored) {
            }
        }
    }

    @Override
    public void shutdown() {
        if (connectionFactory != null) {
            connectionFactory.stop();
            threadPoolManager.shutdown(PoolNameConst.RABBITMQ_POOL);
        }
    }
}